// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//
#pragma once

#include <pollux/type/type.h>

#include <duckdb.hpp> // @manual

namespace kumo::pollux {
    class variant;
}

namespace kumo::pollux::duckdb {
    /// Converts Pollux type to DuckDB type.
    ::duckdb::LogicalType fromPolluxType(const TypePtr &type);

    /// Converts DuckDB type to Pollux type.
    TypePtr toPolluxType(
        ::duckdb::LogicalType type,
        bool fileColumnNamesReadAsLowerCase = false);

    static ::duckdb::timestamp_t polluxTimestampToDuckDB(
        const Timestamp &timestamp) {
        auto micros = timestamp.getSeconds() * 1000000 + timestamp.getNanos() / 1000;
        return ::duckdb::Timestamp::FromEpochMicroSeconds(micros);
    }

    static Timestamp duckdbTimestampToPollux(
        const ::duckdb::timestamp_t &timestamp) {
        auto micros = ::duckdb::Timestamp::GetEpochMicroSeconds(timestamp);

        auto seconds = micros / 1'000'000;
        auto nanoSeconds = (micros % 1'000'000) * 1'000;

        // Make sure nanoseconds are >= 0 even if timestamp represents time before
        // epoch.
        if (nanoSeconds < 0) {
            seconds--;
            nanoSeconds += 1'000'000'000;
        }

        return Timestamp(seconds, nanoSeconds);
    }

    // Converts a duckDB Value (class that holds an arbitrary data type) into
    // Pollux variant.
    variant duckValueToVariant(const ::duckdb::Value &val);

    // Converts duckDB decimal Value into appropriate decimal variant.
    // The duckdb::Value::GetValue() call for decimal type returns a double value.
    // To avoid this, this method uses the duckdb::Value::GetUnsafeValue<int>()
    // method.
    // @param val duckdb decimal value.
    // @return decimal variant.
    variant decimalVariant(const ::duckdb::Value &val);

    // value conversion routines
    template<class T>
    struct DuckNumericConversion {
        typedef T DUCK_TYPE;
        typedef T POLLUX_TYPE;

        static T toDuck(const T &input, ::duckdb::Vector & /* unused */) {
            return input;
        }

        static T toPollux(const T &input) {
            return input;
        }
    };

    struct DuckHugeintConversion {
        typedef ::duckdb::hugeint_t DUCK_TYPE;
        typedef double POLLUX_TYPE;

        static ::duckdb::hugeint_t toDuck(
            const double &input,
            ::duckdb::Vector & /* unused */) {
            return ::duckdb::Hugeint::Convert<double>(input);
        }

        static double toPollux(const ::duckdb::hugeint_t &input) {
            return ::duckdb::Hugeint::Cast<double>(input);
        }
    };

    struct DuckStringConversion {
        typedef ::duckdb::string_t DUCK_TYPE;
        typedef StringView POLLUX_TYPE;

        static ::duckdb::string_t toDuck(
            const StringView &input,
            ::duckdb::Vector &result) {
            return ::duckdb::StringVector::AddString(
                result, input.data(), input.size());
        }

        static StringView toPollux(const ::duckdb::string_t &input) {
            return StringView(input.GetDataUnsafe(), input.GetSize());
        }
    };

    struct DuckBlobConversion {
        typedef ::duckdb::string_t DUCK_TYPE;
        typedef StringView POLLUX_TYPE;

        static ::duckdb::string_t toDuck(
            const StringView &input,
            ::duckdb::Vector &result) {
            return ::duckdb::StringVector::AddStringOrBlob(
                result, input.data(), input.size());
        }

        static StringView toPollux(const ::duckdb::string_t &input) {
            return StringView(input.GetDataUnsafe(), input.GetSize());
        }
    };

    struct DuckInt16DecimalConversion {
        typedef int16_t DUCK_TYPE;
        typedef int64_t POLLUX_TYPE;

        static int16_t toDuck(const int64_t &input, ::duckdb::Vector & /* unused */) {
            return input;
        }

        static int64_t toPollux(const int16_t input) {
            return static_cast<int64_t>(input);
        }
    };

    struct DuckInt32DecimalConversion {
        typedef int32_t DUCK_TYPE;
        typedef int64_t POLLUX_TYPE;

        static int32_t toDuck(const int64_t &input, ::duckdb::Vector & /* unused */) {
            return input;
        }

        static int64_t toPollux(const int32_t input) {
            return static_cast<int64_t>(input);
        }
    };

    struct DuckInt64DecimalConversion {
        typedef int64_t DUCK_TYPE;
        typedef int64_t POLLUX_TYPE;

        static int64_t toDuck(const int64_t &input, ::duckdb::Vector & /* unused */) {
            return input;
        }

        static int64_t toPollux(const int64_t input) {
            return input;
        }
    };

    struct DuckLongDecimalConversion {
        typedef ::duckdb::hugeint_t DUCK_TYPE;
        typedef int128_t POLLUX_TYPE;

        static ::duckdb::hugeint_t toDuck(
            const int128_t &input,
            ::duckdb::Vector & /* unused */) {
            ::duckdb::hugeint_t duckValue;
            duckValue.upper = (input >> 64);
            duckValue.lower = input;
            return duckValue;
        }

        static int128_t toPollux(const ::duckdb::hugeint_t input) {
            return HugeInt::build(input.upper, input.lower);
        }
    };

    struct DuckTimestampConversion {
        typedef ::duckdb::timestamp_t DUCK_TYPE;
        typedef Timestamp POLLUX_TYPE;

        static ::duckdb::timestamp_t toDuck(
            const Timestamp &input,
            ::duckdb::Vector & /* unused */) {
            return polluxTimestampToDuckDB(input);
        }

        static Timestamp toPollux(const ::duckdb::timestamp_t &input) {
            return duckdbTimestampToPollux(input);
        }
    };

    struct DuckDateConversion {
        typedef ::duckdb::date_t DUCK_TYPE;
        typedef int32_t POLLUX_TYPE;

        static ::duckdb::date_t toDuck(
            const int32_t &input,
            ::duckdb::Vector & /* unused */) {
            return ::duckdb::Date::EpochDaysToDate(input);
        }

        static int32_t toPollux(const ::duckdb::date_t &input) {
            return ::duckdb::Date::EpochDays(input);
        }
    };

    /// Returns CREATE TABLE <tableName>(<schema>) DuckDB SQL.
    std::string makeCreateTableSql(
        const std::string &tableName,
        const RowType &rowType);
} // namespace kumo::pollux::duckdb
